package com.wuze.TCC.service.impl;

import com.wuze.TCC.annotation.TCC;
import com.wuze.TCC.pojo.TransactionLog;
import com.wuze.TCC.pojo.TransactionMsg;
import com.wuze.TCC.service.TCCManager;
import com.wuze.TCC.util.DruidUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.rpc.RpcContext;
import java.lang.reflect.Method;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;


/**
 * TCC事务管理器实现类
 * @author wuze
 */
@Slf4j
public class TccManagerImpl implements TCCManager {


    public static Queue<TransactionMsg> queue = new LinkedBlockingQueue<>();

    public static ScheduledExecutorService scheduledExecutorService;

    static{
        scheduledExecutorService = Executors.newScheduledThreadPool(1);
        scheduledExecutorService.scheduleAtFixedRate(()->{
            int size = queue.size();
            if(size == 0){
                return;
            }
            TransactionMsg poll = queue.poll();
//            log.info("{}",poll.getTransId());
            try {
                int i = selectStatus(poll.TransId, DruidUtil.getConnection());
                if(i == 0){
                    queue.add(poll);
                }else if(i == 1){
                    //执行Comfirm方法
                    TCC annotation = poll.getAnnotation();
                    Method method = poll.getMethod();
                    Object target = poll.getTraget();
                    Object[] args = poll.getArgs();
                    Method method1 = target.getClass().getMethod(annotation.confirm(), method.getParameterTypes());

                    for(int a = 0;a<30;a++){
                        try {
                            method1.invoke(target,args);
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                        break;
                    }


                }else{
                    //执行Cancel方法
                    TCC annotation = poll.getAnnotation();
                    Method method = poll.getMethod();
                    Object target = poll.getTraget();
                    Object[] args = poll.getArgs();
                    Method method1 = target.getClass().getMethod(annotation.cancel(), method.getParameterTypes());

                    for(int a = 0;a<30;a++){
                        try {
                            method1.invoke(target,args);
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                        break;
                    }

                }
            } catch (Exception troubles) {
                troubles.printStackTrace();
            }

        },100,20, TimeUnit.MILLISECONDS);

    }

    @Override
    public void doTry(String transId , TCC annotation , Object traget , Method method , Object[] args) throws SQLException, InterruptedException {

        //封装事务日志
        TransactionLog transactionLog = new TransactionLog();
        transactionLog.setTarget_method(traget.getClass().getName()+"."+method.getName());
        transactionLog.setCancel_method(traget.getClass().getName()+"."+annotation.cancel());
        transactionLog.setConfirm_method(traget.getClass().getName()+"."+annotation.confirm());
        transactionLog.setTarget_class(traget.getClass().getName());
        transactionLog.setRetried_count(0);
        transactionLog.setCreate_time(new Date());
        transactionLog.setLast_time(new Date());
        // 0:trying 1:confirm 2:cancel
        transactionLog.setStatus(0);
        Connection connection = DruidUtil.getConnection();
        //判断是否是服务发起者
        if(annotation.isInitiator()){
            //服务发起者
            transactionLog.setRole(0);
            transactionLog.setTrans_id(transId);
            saveTryTransactionLogs(transId,transactionLog,connection);
            RpcContext.getContext().setAttachment("TransId", transId);

        }else{
            //服务提供者
            transactionLog.setRole(1);
            transactionLog.setTrans_id(transId);
            saveTryTransactionLogs(transId,transactionLog,connection);
        }
        TransactionMsg transactionMsg = new TransactionMsg();
        transactionMsg.setMethod(method);
        transactionMsg.setTransId(transId);
        transactionMsg.setTraget(traget);
        transactionMsg.setAnnotation(annotation);
        transactionMsg.setArgs(args);
        queue.add(transactionMsg);

    }

    @Override
    public void doComfirm(String transId , TCC annotation , Object traget , Method method , Object[] args) throws SQLException {
        updateTransactionLogs(transId,DruidUtil.getConnection(), 1 , annotation);
    }

    @Override
    public void doCancel(String transId , TCC annotation , Object traget , Method method , Object[] args) throws SQLException {
        updateTransactionLogs(transId,DruidUtil.getConnection(), 2 , annotation);
    }


    public static void saveTryTransactionLogs(String TransId ,
                                              TransactionLog transactionLog,
                                              Connection connection) throws SQLException {


        Statement statement = null;
        try {
            String target_class = transactionLog.getTarget_class();
            String target_method = transactionLog.getTarget_method();
            String confirm_method = transactionLog.getConfirm_method();
            String cancel_method = transactionLog.getCancel_method();
            int retried_count = transactionLog.getRetried_count();
            String create_time = transactionLog.getCreate_time().toString();
            String last_time = transactionLog.getLast_time().toString();
            int status = transactionLog.getStatus();
            int role  = transactionLog.getRole();

            String sql = "";
            if(role == 0){
                sql = "INSERT INTO comsumer_log VALUES (\""+TransId+"\",\""+target_class+"\",\""+target_method+"\",\""+confirm_method+"\",\""+cancel_method+"\","+retried_count+",'"+create_time+"','"+last_time+"',"+status+","+role+")";

            }else{
                sql = "INSERT INTO provider_log VALUES (\""+TransId+"\",\""+target_class+"\",\""+target_method+"\",\""+confirm_method+"\",\""+cancel_method+"\","+retried_count+",'"+create_time+"','"+last_time+"',"+status+","+role+")";

            }
            log.info("Try阶段====>");
            statement = connection.createStatement();
            int i = statement.executeUpdate(sql);
        }catch (Exception e){

        }finally {
            statement.close();
            connection.close();
        }

    }

    public static void updateTransactionLogs(String TransId ,
                                              Connection connection,
                                              int status,
                                             TCC annotation) throws SQLException {

        Statement statement = null;
        try {
            String time = new Date().toString();
            String sql="";
            if(annotation.isInitiator()){
                sql = "UPDATE comsumer_log SET `status`="+status+",last_time=\""+time+"\" WHERE trans_id =\""+TransId+"\"";
            }else{
                sql = "UPDATE provider_log SET `status`="+status+",last_time=\""+time+"\" WHERE trans_id =\""+TransId+"\"";
            }
            statement = connection.createStatement();
            int i = statement.executeUpdate(sql);


        }catch (Exception e){
            e.printStackTrace();
        }finally {
            statement.close();
            connection.close();
        }


    }


    public static int selectStatus(String TransId ,
                                    Connection connection) throws SQLException {

        Statement statement = null;
        try {

            String s1 = "SELECT `status` FROM comsumer_log WHERE trans_id=\""+TransId+"\"";
            String s2 = "SELECT `status` FROM provider_log WHERE trans_id=\""+TransId+"\"";

            statement = connection.createStatement();
            ResultSet resultSet1 = statement.executeQuery(s1);

            if (!resultSet1.next()) {
                return 0;
            }



            int status1 = resultSet1.getInt("status");

            ResultSet resultSet2 = statement.executeQuery(s2);
            if (!resultSet2.next()) {
                return 0;
            }
            int status2 = resultSet2.getInt("status");

            if(status1 == 2 || status2 == 2){
                return 2;
            }

            if(status1 == 1 && status2 == 1){
                return 1;
            }

            return 0;

        }catch (Exception e){
            return 0;
        }finally {
            connection.close();
            statement.close();
        }

    }




}
